import org.apache.spark._
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming._

import scala.collection.mutable
/**
  * Created by hadoop on 17-6-2.
  */

object RddQueueStreaming {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("RddQueueStreaming").setMaster("local[2]")
    val sc = new SparkContext(conf)
    val ssc = new StreamingContext(sc,Seconds(20))
    val rddQueue = new mutable.SynchronizedQueue[RDD[Int]]()
    val queueStream = ssc.queueStream(rddQueue)
    val reducedStream = queueStream.map(r=>(r%10,1)).reduceByKey(_+_)
    reducedStream.print()
    ssc.start()
    for(i <- 1 to 10){
      rddQueue += sc.makeRDD(1 to 100 ,2)
      Thread.sleep(1000)
    }
    ssc.stop()
  }

}
